single pub sub lock fix - #28
Conversation
|
Thank you for your contribution to @imqueue. Before this pull request can be merged, please read the @imqueue Contribution Terms and sign them by posting the following comment exactly: I have read the CLA Document and I hereby sign the CLA You can retrigger this bot by commenting recheck in this Pull Request. Posted by the CLA Assistant Lite bot. |
There was a problem hiding this comment.
Pull request overview
This PR addresses a single-listener edge case where a PgIpLock can be acquired later by the retry timer after the original listen() call has already returned, leaving the lock held but the intended LISTEN never executed. It introduces a late-acquire callback to complete the “second half” of the work (subscribing) when takeover happens asynchronously.
Changes:
- Add
onAcquire()to theAnyLockcontract and implement it inPgIpLockandNoLock. - Refactor
PgPubSub.listen()to delegate the actualLISTEN+emit into a newsubscribe()method, enabling late-takeover subscription without re-acquiring the lock. - Add tests covering
onAcquire()behavior inPgIpLock.
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| test/PgIpLock.spec.ts | Adds tests for the new late-acquire callback behavior. |
| src/types/AnyLock.ts | Extends lock interface with a new onAcquire() hook and docs. |
| src/PgPubSub.ts | Uses subscribe() for LISTEN+emit, wires late-takeover handler to complete subscription. |
| src/PgIpLock.ts | Implements onAcquire() and adds timer-driven acquire path that invokes it on success. |
| src/NoLock.ts | Implements onAcquire() as a no-op for the no-lock strategy. |
Comments suppressed due to low confidence (1)
src/PgIpLock.ts:281
- Timer-driven retries can overlap if acquire() takes longer than acquireInterval, which risks multiple handler calls and also racing this.acquired updates. Use a simple in-flight guard + finally to ensure the guard is always cleared even if the handler throws.
private async retryAcquire(): Promise<void> {
if (this.acquired) {
return;
}
try {
if (await this.acquire()) {
this.acquireHandler?.(this.publicChannel);
}
} catch (err) {
this.options.logger.error(err);
}
}
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| lock.onRelease(chan => this.listen(chan)); | ||
| lock.onAcquire(chan => | ||
| this.subscribe(chan).catch(err => this.logger.error(err)), | ||
| ); |
|
I have read the CLA Document and I hereby sign the CLA |
2c88f1f to
43a8d57
Compare
What does this PR do?
Type of change
Checklist
npm test).Contribution terms (required)
I grant the project owner the right to license my contribution
commercially, royalty-free, my contribution stays available under
GPL-3.0, I keep my copyright, and I understand I will receive no fee for
it. If I did not agree, I would not be submitting this contribution.